package yuekao6.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import yuekao6.entity.TransportTask;
import yuekao6.util.KafkaUtil;

import java.text.SimpleDateFormat;
import java.util.TimeZone;

public class TranSportData {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //（4）实时数仓DWD层-业务从原始业务数据库的transport_task表中提取运单完成操作，写入Kafka指定主题。环境准备，从 Kafka tms_ods 主题读取数据，
        // 筛选运输完成数据，处理运输完成数据，获取修改后的数据并转换数据类型，补全时间戳字段，补全运输时长字段，处理时区问题，脱敏，
        // 写出到 Kafka tms_dwd_trans_trans_finish 主题物流域运输完成事实主题（5分）
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("tms_ods"));

//        streamSource.print();
        SingleOutputStreamOperator<String> filterDS = streamSource.filter(x -> JSON.parseObject(x).getString("table").equals("transport_task")&&JSON.parseObject(x).getJSONObject("data").getString("status").equals("67004"));
//        filterDS.print();

        SingleOutputStreamOperator<String> dataDS = filterDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                TransportTask data = JSON.parseObject(JSON.parseObject(s).getJSONObject("data").toJSONString(), TransportTask.class);
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));
                long time = sdf.parse(data.getCreate_time()).getTime();
                //时间戳字段
                data.setCreate_ts(time);
                //运输时长
                data.setTransport_time(data.getActual_start_time() - data.getActual_end_time());

                //脱敏
                data.setDriver1_name(data.getDriver1_name() != null ? data.getDriver1_name().substring(0, 1) + "**" : data.getDriver1_name());

                data.setDriver2_name(data.getDriver2_name() != null ? data.getDriver2_name().substring(0, 1) + "**" : data.getDriver2_name());

                data.setTruck_no(data.getTruck_no() != null ? data.getTruck_no().substring(0, 3) + "**" : data.getTruck_no());
                return JSONObject.toJSONString(data);
            }
        });
        dataDS.print();
        dataDS.addSink(KafkaUtil.kafkaSink("tms_dwd_trans_trans_finish"));

        env.execute();
    }
}
